﻿#include <iostream>
#include <iomanip>
#include <set>

#include <mqtt_server_cpp.hpp>

#include <boost/lexical_cast.hpp>
#include <boost/multi_index_container.hpp>
#include <boost/multi_index/ordered_index.hpp>
#include <boost/multi_index/member.hpp>
#include <boost/multi_index/composite_key.hpp>

#include "gtest/gtest.h"
#include "log.h"

#include "net/mqtt/mqtt_no_ts_server.h"

// 多索引容器
namespace mi = boost::multi_index;

using con_t = MQTT_NS::server<>::endpoint_t;
using con_sp_t = std::shared_ptr<con_t>;

struct sub_con {
    sub_con(MQTT_NS::buffer topic, con_sp_t con, MQTT_NS::qos qos_value)
            : topic(std::move(topic)), con(std::move(con)), qos_value(qos_value) {}

    MQTT_NS::buffer topic;
    con_sp_t con;
    MQTT_NS::qos qos_value;
};

// 如果要把sub_con某个属性字段设置为搜索引擎，则需要定义用于排序的空结构体对象
struct tag_topic {
};
struct tag_con {
};
struct tag_con_topic {
};

//  BOOST_MULTI_INDEX_MEMBER(索引源的对象或者结构体类型, index_item的类型 ,  index_item ),
// 定义一个multi_index_container（多索引容器）
using mi_sub_con = mi::multi_index_container<
        sub_con,
        mi::indexed_by<
                mi::ordered_unique<
                        mi::tag<tag_con_topic>,
                        mi::composite_key<
                                sub_con,
                                BOOST_MULTI_INDEX_MEMBER(sub_con, con_sp_t, con),
                                BOOST_MULTI_INDEX_MEMBER(sub_con, MQTT_NS::buffer, topic)
                        >
                >,//con 与 topic 联合主键
                mi::ordered_non_unique<
                        mi::tag<tag_topic>,
                        BOOST_MULTI_INDEX_MEMBER(sub_con, MQTT_NS::buffer, topic)
                >,//topic 非唯一索引
                mi::ordered_non_unique<
                        mi::tag<tag_con>,
                        BOOST_MULTI_INDEX_MEMBER(sub_con, con_sp_t, con)
                > //con  非唯一索引
        >
>;


inline void close_proc(std::set<con_sp_t> &cons, mi_sub_con &subs, con_sp_t const &con) {
    cons.erase(con);

    auto &idx = subs.get<tag_con>();
    auto r = idx.equal_range(con);
    idx.erase(r.first, r.second);
}

TEST(mqtt, mqtt_server) {

    MQTT_NS::setup_log();

    boost::asio::io_context ioc;

    auto s = MQTT_NS::server<>(
            boost::asio::ip::tcp::endpoint(
                    boost::asio::ip::tcp::v4(),
                    boost::lexical_cast<std::uint16_t>("1883")
            ),
            ioc
    );

    s.set_error_handler(
            [](MQTT_NS::error_code ec)
            {
                LOG_INFO << "error: " << ec.message();
            }
    );

    std::set<con_sp_t> connections;
    mi_sub_con subs;

    // 接收连接的处理
    s.set_accept_handler(
            [&connections, &subs](con_sp_t spep)
            {
                auto &ep = *spep;
                std::weak_ptr<con_t> wp(spep);

                using packet_id_t = typename std::remove_reference_t<decltype(ep)>::packet_id_t;
                LOG_INFO << "accept";

                // Pass spep to keep lifetime.
                // It makes sure wp.lock() never return nullptr in the handlers below
                // including close_handler and error_handler.
                ep.start_session(std::move(spep));

                // 正常关闭
                ep.set_close_handler([&connections, &subs, wp]()
                                     {
                                         LOG_INFO << "[server] closed.";
                                         auto sp = wp.lock();
                                         BOOST_ASSERT(sp);
                                         close_proc(connections, subs, sp);
                                     });

                // 超时关闭 或者 对方强制关闭 处理
                ep.set_error_handler([&connections, &subs, wp](MQTT_NS::error_code ec)
                                     {
                                         LOG_INFO << "[server] error: " << ec.message();
                                         auto sp = wp.lock();
                                         BOOST_ASSERT(sp);
                                         close_proc(connections, subs, sp);
                                     });

                // 权限校验
                // set MQTT level handlers
                ep.set_connect_handler(
                        [&connections, wp](MQTT_NS::buffer client_id, MQTT_NS::optional<MQTT_NS::buffer> username,
                                           MQTT_NS::optional<MQTT_NS::buffer> password,
                                           MQTT_NS::optional<MQTT_NS::will>,
                                           bool clean_session,
                                           std::uint16_t keep_alive)
                        {
                            using namespace MQTT_NS::literals;
                            LOG_INFO << "[server] client_id    : " << client_id;
                            LOG_INFO << "[server] username     : " << (username ? username.value() : "none"_mb);
                            LOG_INFO << "[server] password     : " << (password ? password.value() : "none"_mb);
                            LOG_INFO << "[server] clean_session: " << std::boolalpha << clean_session;
                            LOG_INFO << "[server] keep_alive   : " << keep_alive;
                            auto sp = wp.lock();
                            BOOST_ASSERT(sp);
                            connections.insert(sp);
                            sp->connack(false, MQTT_NS::connect_return_code::accepted);
                            return true;
                        }
                );

                // 关闭连接
                ep.set_disconnect_handler([&connections, &subs, wp]()
                                          {
                                              LOG_INFO << "[server] disconnect received.";
                                              auto sp = wp.lock();
                                              BOOST_ASSERT(sp);
                                              close_proc(connections, subs, sp);
                                          });

                //控制报文之发布消息（PUBLISH）
                // 发布确认（PUBACK）
                // 发布收到（PUBREC）
                // 发布释放（PUBREL）
                // 发布完成（PUBCOMP）

                // Qos 1级响应报文
                ep.set_puback_handler([](packet_id_t packet_id)
                                      {
                                          LOG_INFO << "[server] puback received. packet_id: " << packet_id;
                                          return true;
                                      });

                // Qos 2级响应报文
                ep.set_pubrec_handler([](packet_id_t packet_id)
                                      {
                                          LOG_INFO << "[server] pubrec received. packet_id: " << packet_id;
                                          return true;
                                      });

                // pu brel报文是对pu brec报文的响应
                ep.set_pubrel_handler([](packet_id_t packet_id)
                                      {
                                          LOG_INFO << "[server] pubrel received. packet_id: " << packet_id;
                                          return true;
                                      });

                ep.set_pubcomp_handler(
                        []
                                (packet_id_t packet_id)
                        {
                            LOG_INFO << "[server] pubcomp received. packet_id: " << packet_id;
                            return true;
                        });

                // 收到消息的处理
                ep.set_publish_handler(
                        [&subs](MQTT_NS::optional<packet_id_t> packet_id,
                                MQTT_NS::publish_options pubopts,
                                MQTT_NS::buffer topic_name,
                                MQTT_NS::buffer contents)
                        {
                            LOG_INFO << "[server] publish received."
                                     << " dup: " << pubopts.get_dup()
                                     << " qos: " << pubopts.get_qos()
                                     << " retain: " << pubopts.get_retain()
                                     << " topic_name: " << topic_name
                                     << " contents: " << contents
                                     << "packet_id:" << (packet_id ? *packet_id:0);

                            // 获取tag_topic所有的索引
                            auto const &idx = subs.get<tag_topic>();

                            // 获取有相同topic的集合，遍历订阅 发送消息
                            auto r = idx.equal_range(topic_name);
                            for (; r.first != r.second; ++r.first)
                            {
                                r.first->con->publish(
                                        topic_name,
                                        contents,
                                        std::min(r.first->qos_value, pubopts.get_qos())
                                );
                            }
                            return true;
                        });

                // 订阅
                ep.set_subscribe_handler(
                        [&subs, wp](packet_id_t packet_id, std::vector<MQTT_NS::subscribe_entry> entries)
                        {
                            LOG_INFO << "[server] subscribe received. packet_id: " << packet_id;
                            std::vector<MQTT_NS::suback_return_code> res;
                            res.reserve(entries.size());
                            auto sp = wp.lock();
                            BOOST_ASSERT(sp);
                            for (auto const &e: entries)
                            {
                                LOG_INFO << "[server] topic_filter: " << e.topic_filter
                                         << " qos: " << e.subopts.get_qos();
                                res.emplace_back(MQTT_NS::qos_to_suback_return_code(e.subopts.get_qos()));
                                subs.emplace(std::move(e.topic_filter), sp, e.subopts.get_qos());
                            }
                            sp->suback(packet_id, res);
                            return true;
                        }
                );

                // 取消订阅
                ep.set_unsubscribe_handler(
                        [&subs, wp](packet_id_t packet_id,
                                    std::vector<MQTT_NS::unsubscribe_entry> entries)
                        {
                            LOG_INFO << "[server] unsubscribe received. packet_id: " << packet_id;
                            auto sp = wp.lock();
                            for (auto const &e: entries)
                            {
                                auto it = subs.find(std::make_tuple(sp, e.topic_filter));
                                if (it != subs.end())
                                {
                                    subs.erase(it);
                                }
                            }
                            BOOST_ASSERT(sp);
                            sp->unsuback(packet_id);
                            return true;
                        }
                );
            }
    );

    s.listen();

    ioc.run();
}

TEST(mqtt, mqtt_server_class) {
    MqttNoTsServer mqtt;
    mqtt.init();
    mqtt.run();
}
/*


#include <mqtt/config.hpp>
#include <mqtt/setup_log.hpp>
#include <mqtt/broker/broker.hpp>
#include <boost/program_options.hpp>
#include <boost/format.hpp>

#include <fstream>
#include <iostream>
#include <iomanip>
#include <algorithm>

namespace as = boost::asio;

using con_t = MQTT_NS::server<>::endpoint_t;
using con_sp_t = std::shared_ptr<con_t>;

class server_no_tls {
public:
    server_no_tls(
            as::io_context& ioc_accept,
            std::function<as::io_context&()> ioc_con_getter,
            MQTT_NS::broker::broker_t& b,
            uint16_t port
    )
            : server_(
            as::ip::tcp::endpoint(
                    as::ip::tcp::v4(), port
            ),
            ioc_accept,
            MQTT_NS::force_move(ioc_con_getter),
            [](auto& acceptor) {
                acceptor.set_option(as::ip::tcp::acceptor::reuse_address(true));
            }
    ), b_(b) {
        server_.set_error_handler(
                [](MQTT_NS::error_code */
/*ec*//*
) {
                }
        );

        server_.set_accept_handler(
                [&](con_sp_t spep) {
                    b_.handle_accept(MQTT_NS::force_move(spep));
                }
        );
    }

    void listen() {
        server_.listen();
    }

    MQTT_NS::broker::broker_t& broker() const {
        return b_;
    }

    void close() {
        server_.close();
    }

private:
    MQTT_NS::server<> server_;
    MQTT_NS::broker::broker_t& b_;
};

#if defined(MQTT_USE_TLS)

inline
bool verify_certificate(
    std::string const& verify_field,
    bool preverified,
    boost::asio::ssl::verify_context& ctx,
    std::shared_ptr<MQTT_NS::optional<std::string>> const& username) {
    if (!preverified) return false;
    int error = X509_STORE_CTX_get_error(ctx.native_handle());
    if (error != X509_V_OK) {
        int depth = X509_STORE_CTX_get_error_depth(ctx.native_handle());
        MQTT_LOG("mqtt_broker", error)
            << "Certificate validation failed, depth: " << depth
            << ", message: " << X509_verify_cert_error_string(error);
        return false;
    }

    X509* cert = X509_STORE_CTX_get_current_cert(ctx.native_handle());
    X509_NAME* name = X509_get_subject_name(cert);

    std::string verify_field_value;
    auto obj = std::unique_ptr<
        ASN1_OBJECT,
        decltype(&ASN1_OBJECT_free)
    >(
        OBJ_txt2obj(verify_field.c_str(), 0),
        &ASN1_OBJECT_free
    );
    if (obj) { // return nullptr if error
        verify_field_value.resize(MQTT_NS::broker::max_cn_size);
        auto size = X509_NAME_get_text_by_OBJ(
            name,
            obj.get(),
            &verify_field_value[0],
            static_cast<int>(verify_field_value.size())
        );
        // Size equals -1 if field is not found, otherwise, length of value
        verify_field_value.resize(static_cast<std::size_t>(std::max(size, 0)));
        MQTT_LOG("mqtt_broker", info) << "[clicrt] " << verify_field << ":" << verify_field_value;
        *username = verify_field_value;
    }
    return true;
}

class server_tls {
public:
    server_tls(
        as::io_context& ioc_accept,
        std::function<as::io_context&()> ioc_con_getter,
        boost::asio::ssl::context&& ctx,
        MQTT_NS::broker::broker_t& b,
        uint16_t port,
        std::string const &verify_field
    )
        : server_(
            as::ip::tcp::endpoint(
                as::ip::tcp::v4(), port
            ),
            MQTT_NS::force_move(ctx),
            ioc_accept,
            MQTT_NS::force_move(ioc_con_getter),
            [](auto& acceptor) {
                acceptor.set_option(as::ip::tcp::acceptor::reuse_address(true));
            }
        ), b_(b) {
        server_.set_error_handler(
            [](MQTT_NS::error_code */
/*ec*//*
) {
            }
        );

        server_.set_verify_callback(
            [&verify_field]
            (
                bool preverified,
                boost::asio::ssl::verify_context& ctx,
                std::shared_ptr<MQTT_NS::optional<std::string>> const& username
            ) {
                return verify_certificate(verify_field, preverified, ctx, username);
            }
        );

        server_.set_accept_handler(
            [this](std::shared_ptr<MQTT_NS::server_tls<>::endpoint_t> spep) {
                b_.handle_accept(MQTT_NS::force_move(spep));
            }
        );
    }

    void listen() {
        server_.listen();
    }

    MQTT_NS::broker::broker_t& broker() const {
        return b_;
    }

    void close() {
        server_.close();
    }

    */
/**
     * @brief Get boost asio ssl context.
     * @return ssl context
     *//*

    MQTT_NS::tls::context& get_ssl_context() {
        return server_.get_ssl_context();
    }

    */
/**
     * @brief Get boost asio ssl context.
     * @return ssl context
     *//*

    MQTT_NS::tls::context const& get_ssl_context() const {
        return server_.get_ssl_context();
    }

private:
    MQTT_NS::server_tls<> server_;
    MQTT_NS::broker::broker_t& b_;
};

#endif // defined(MQTT_USE_TLS)

#if defined(MQTT_USE_WS)

class server_no_tls_ws {
public:
    server_no_tls_ws(
        as::io_context& ioc_accept,
        std::function<as::io_context&()> ioc_con_getter,
        MQTT_NS::broker::broker_t& b,
        uint16_t port)
        : server_(
            as::ip::tcp::endpoint(
                as::ip::tcp::v4(), port
            ),
            ioc_accept,
            MQTT_NS::force_move(ioc_con_getter),
            [](auto& acceptor) {
                acceptor.set_option(as::ip::tcp::acceptor::reuse_address(true));
            }
        ), b_(b) {
        server_.set_error_handler(
            [](MQTT_NS::error_code */
/*ec*//*
) {
            }
        );

        server_.set_accept_handler(
            [this](std::shared_ptr<MQTT_NS::server_ws<>::endpoint_t> spep) {
                b_.handle_accept(MQTT_NS::force_move(spep));
            }
        );
    }

    void listen() {
        server_.listen();
    }

    MQTT_NS::broker::broker_t& broker() const {
        return b_;
    }

    void close() {
        server_.close();
    }

private:
    MQTT_NS::server_ws<> server_;
    MQTT_NS::broker::broker_t& b_;
};

#if defined(MQTT_USE_TLS)

class server_tls_ws {
public:
    server_tls_ws(
        as::io_context& ioc_accept,
        std::function<as::io_context&()> ioc_con_getter,
        boost::asio::ssl::context&& ctx,
        MQTT_NS::broker::broker_t& b,
        uint16_t port,
        std::string const &verify_field
    )
        : server_(
            as::ip::tcp::endpoint(
                as::ip::tcp::v4(),
                port
            ),
            MQTT_NS::force_move(ctx),
            ioc_accept,
            MQTT_NS::force_move(ioc_con_getter),
            [](auto& acceptor) {
                acceptor.set_option(as::ip::tcp::acceptor::reuse_address(true));
            }
        ), b_(b) {
        server_.set_error_handler(
            [](MQTT_NS::error_code */
/*ec*//*
) {
            }
        );

        server_.set_verify_callback(
            [&verify_field]
            (
                bool preverified,
                boost::asio::ssl::verify_context& ctx,
                std::shared_ptr<MQTT_NS::optional<std::string>> const& username
            ) {
                return verify_certificate(verify_field, preverified, ctx, username);
            }
        );

        server_.set_accept_handler(
            [this](std::shared_ptr<MQTT_NS::server_tls_ws<>::endpoint_t> spep) {
                b_.handle_accept(MQTT_NS::force_move(spep));
            }
        );
    }

    void listen() {
        server_.listen();
    }

    MQTT_NS::broker::broker_t& broker() const {
        return b_;
    }

    void close() {
        server_.close();
    }

    */
/**
     * @brief Get boost asio ssl context.
     * @return ssl context
     *//*

    MQTT_NS::tls::context& get_ssl_context() {
        return server_.get_ssl_context();
    }

    */
/**
     * @brief Get boost asio ssl context.
     * @return ssl context
     *//*

    MQTT_NS::tls::context const& get_ssl_context() const {
        return server_.get_ssl_context();
    }

private:
    MQTT_NS::server_tls_ws<> server_;
    MQTT_NS::broker::broker_t& b_;
};

#endif // defined(MQTT_USE_TLS)
#endif // defined(MQTT_USE_WS)


#if defined(MQTT_USE_TLS)
as::ssl::context init_ctx()
{
    as::ssl::context ctx(as::ssl::context::tlsv12);
    ctx.set_options(
        as::ssl::context::default_workarounds |
        as::ssl::context::single_dh_use);
    return ctx;
}

template<typename Server>
void reload_ctx(Server& server,
                as::steady_timer& reload_timer,
                std::string const& certificate_filename,
                std::string const& key_filename,
                MQTT_NS::optional<std::string> const& verify_file,
                unsigned int certificate_reload_interval,
                char const* name,
                bool first_load = true) {
    MQTT_LOG("mqtt_broker", info) << "Reloading certificates for server " << name;

    if (certificate_reload_interval > 0) {
        reload_timer.expires_after(std::chrono::hours(certificate_reload_interval));
        reload_timer.async_wait(
            [
                &server,
                &reload_timer,
                certificate_filename,
                key_filename,
                certificate_reload_interval,
                verify_file,
                name
            ]
            (boost::system::error_code const& e) {
                BOOST_ASSERT(!e || e == as::error::operation_aborted);
                if (!e) {
                    reload_ctx(
                        server,
                        reload_timer,
                        certificate_filename,
                        key_filename,
                        verify_file,
                        certificate_reload_interval,
                        name,
                        false
                    );
                }
            }
        );
    }

    auto context = init_ctx();

    boost::system::error_code ec;
    context.use_certificate_chain_file(certificate_filename, ec);
    if (ec) {
        auto message = "Failed to load certificate file: " + ec.message();
        if (first_load) {
            throw std::runtime_error(message);
        }

        MQTT_LOG("mqtt_broker", warning) << message;
        return;
    }

    context.use_private_key_file(key_filename, as::ssl::context::pem, ec);
    if (ec) {
        auto message = "Failed to load private key file: " + ec.message();
        if (first_load) {
            throw std::runtime_error(message);
        }

        MQTT_LOG("mqtt_broker", warning) << message;
        return;
    }

    if (verify_file) {
        context.load_verify_file(*verify_file);
    }

    server.get_ssl_context() = MQTT_NS::force_move(context);
}

template<typename Server>
void load_ctx(
    Server& server,
    as::steady_timer& reload_timer,
    boost::program_options::variables_map const& vm,
    char const* name
) {
    if (vm.count("certificate") == 0 && vm.count("private_key") == 0) {
        throw std::runtime_error("TLS requested but certificate and/or private_key not specified");
    }

    MQTT_NS::optional<std::string> verify_file;
    if (vm.count("verify_file")) {
        verify_file = vm["verify_file"].as<std::string>();
    }
    reload_ctx(
        server,
        reload_timer,
        vm["certificate"].as<std::string>(),
        vm["private_key"].as<std::string>(),
        verify_file,
        vm["certificate_reload_interval"].as<unsigned int>(),
        name, true
    );
}
#endif // defined(MQTT_USE_TLS)

void run_broker(boost::program_options::variables_map const& vm) {
    try {
        as::io_context timer_ioc;
        MQTT_NS::broker::broker_t b(timer_ioc);

        auto num_of_iocs =
                [&] () -> std::size_t {
                    if (vm.count("iocs")) {
                        return vm["iocs"].as<std::size_t>();
                    }
                    return 1;
                } ();
        if (num_of_iocs == 0) {
            num_of_iocs = std::thread::hardware_concurrency();
            MQTT_LOG("mqtt_broker", info)
                    << "iocs set to auto decide (0). Automatically set to " << num_of_iocs;
        }

        auto threads_per_ioc =
                [&] () -> std::size_t {
                    if (vm.count("threads_per_ioc")) {
                        return vm["threads_per_ioc"].as<std::size_t>();
                    }
                    return 1;
                } ();
        if (threads_per_ioc == 0) {
            threads_per_ioc = std::min(std::size_t(std::thread::hardware_concurrency()), std::size_t(4));
            MQTT_LOG("mqtt_broker", info)
                    << "threads_per_ioc set to auto decide (0). Automatically set to "
                    << threads_per_ioc;
        }

        MQTT_LOG("mqtt_broker", info)
                << "iocs:" << num_of_iocs
                << " threads_per_ioc:" << threads_per_ioc
                << " total threads:" << num_of_iocs * threads_per_ioc;

        if (vm.count("auth_file")) {
            std::string auth_file = vm["auth_file"].as<std::string>();
            if (!auth_file.empty()) {
                MQTT_LOG("mqtt_broker", info)
                        << "auth_file:" << auth_file;

                std::ifstream input(auth_file);

                if (input) {
                    MQTT_NS::broker::security security;
                    security.load_json(input);
                    b.set_security(MQTT_NS::force_move(security));
                }
                else {
                    MQTT_LOG("mqtt_broker", error)
                            << "Authorization file '"
                            << auth_file
                            << "' not found,  broker doesn't use authorization file.";
                }
            }
        }

        as::io_context accept_ioc;

        std::mutex mtx_con_iocs;
        std::vector<as::io_context> con_iocs(num_of_iocs);
        BOOST_ASSERT(!con_iocs.empty());

        std::vector<
                as::executor_work_guard<
                        as::io_context::executor_type
                >
        > guard_con_iocs;
        guard_con_iocs.reserve(con_iocs.size());
        for (auto& con_ioc : con_iocs) {
            guard_con_iocs.emplace_back(con_ioc.get_executor());
        }

        auto con_iocs_it = con_iocs.begin();

        auto con_ioc_getter =
                [&mtx_con_iocs, &con_iocs, &con_iocs_it]() -> as::io_context& {
                    std::lock_guard<std::mutex> g{mtx_con_iocs};
                    auto& ret = *con_iocs_it++;
                    if (con_iocs_it == con_iocs.end()) con_iocs_it = con_iocs.begin();
                    return ret;
                };

        MQTT_NS::optional<server_no_tls> s;
        if (vm.count("tcp.port")) {
            s.emplace(
                    accept_ioc,
                    con_ioc_getter,
                    b,
                    vm["tcp.port"].as<std::uint16_t>()
            );
            s->listen();
        }

#if defined(MQTT_USE_WS)
        MQTT_NS::optional<server_no_tls_ws> s_ws;
        if (vm.count("ws.port")) {
            s_ws.emplace(
                accept_ioc,
                con_ioc_getter,
                b,
                vm["ws.port"].as<std::uint16_t>()
            );
            s_ws->listen();
        }
#endif // defined(MQTT_USE_WS)

#if defined(MQTT_USE_TLS)
        MQTT_NS::optional<server_tls> s_tls;
        MQTT_NS::optional<as::steady_timer> s_lts_timer;

        auto verify_field_obj =
            std::unique_ptr<ASN1_OBJECT, decltype(&ASN1_OBJECT_free)>(
                OBJ_txt2obj(vm["verify_field"].as<std::string>().c_str(), 0),
                &ASN1_OBJECT_free
            );
        if (!verify_field_obj) {
            throw std::runtime_error(
                "An invalid verify field was specified: " +
                vm["verify_field"].as<std::string>()
            );
        }
        if (vm.count("tls.port")) {
            s_tls.emplace(
                accept_ioc,
                con_ioc_getter,
                init_ctx(),
                b,
                vm["tls.port"].as<std::uint16_t>(),
                vm["verify_field"].as<std::string>()
            );
            s_lts_timer.emplace(accept_ioc);
            load_ctx(s_tls.value(), s_lts_timer.value(), vm, "TLS");
            s_tls->listen();
        }
#endif // defined(MQTT_USE_TLS)

#if defined(MQTT_USE_TLS) && defined(MQTT_USE_WS)
        MQTT_NS::optional<server_tls_ws> s_tls_ws;
        MQTT_NS::optional<as::steady_timer> s_tls_ws_timer;

        if (vm.count("wss.port")) {
            s_tls_ws.emplace(
                accept_ioc,
                con_ioc_getter,
                init_ctx(),
                b,
                vm["wss.port"].as<std::uint16_t>(),
                vm["verify_field"].as<std::string>()
            );
            s_tls_ws_timer.emplace(accept_ioc);
            load_ctx(s_tls_ws.value(), s_tls_ws_timer.value(), vm, "WSS");
            s_tls_ws->listen();
        }
#endif // defined(MQTT_USE_TLS) && defined(MQTT_USE_WS)

        std::thread th_accept {
                [&accept_ioc] {
                    accept_ioc.run();
                    MQTT_LOG("mqtt_broker", trace) << "accept_ioc.run() finished";
                }
        };

        as::executor_work_guard<
                as::io_context::executor_type
        > guard_timer_ioc(timer_ioc.get_executor());

        std::thread th_timer {
                [&timer_ioc] {
                    timer_ioc.run();
                    MQTT_LOG("mqtt_broker", trace) << "timer_ioc.run() finished";
                }
        };
        std::vector<std::thread> ts;
        ts.reserve(num_of_iocs * threads_per_ioc);
        for (auto& con_ioc : con_iocs) {
            for (std::size_t i = 0; i != threads_per_ioc; ++i) {
                ts.emplace_back(
                        [&con_ioc] {
                            con_ioc.run();
                            MQTT_LOG("mqtt_broker", trace) << "con_ioc.run() finished";
                        }
                );
            }
        }

        as::io_context ioc_signal;
        as::signal_set signals{ioc_signal, SIGINT, SIGTERM};
        signals.async_wait(
                [] (
                        boost::system::error_code const& ec,
                        int num
                ) {
                    if (!ec) {
                        MQTT_LOG("mqtt_broker", trace)
                                << "Signal " << num << " received. exit program";
                        exit(-1);
                    }
                }
        );
        std::thread th_signal {
                [&] {
                    ioc_signal.run();
                }
        };

        th_accept.join();
        MQTT_LOG("mqtt_broker", trace) << "th_accept joined";

        for (auto& g : guard_con_iocs) g.reset();
        for (auto& t : ts) t.join();
        MQTT_LOG("mqtt_broker", trace) << "ts joined";

        guard_timer_ioc.reset();
        th_timer.join();
        MQTT_LOG("mqtt_broker", trace) << "th_timer joined";

        signals.cancel();
        th_signal.join();
        MQTT_LOG("mqtt_broker", trace) << "th_signal joined";

    } catch(std::exception &e) {
        MQTT_LOG("mqtt_broker", error) << e.what();
    }
}

int main(int argc, char **argv) {
    try {
        boost::program_options::options_description desc;

        boost::program_options::options_description general_desc("General options");
        general_desc.add_options()
                ("help", "produce help message")
                (
                        "cfg",
                        boost::program_options::value<std::string>()->default_value("broker.conf"),
                        "Load configuration file"
                )
                (
                        "iocs",
                        boost::program_options::value<std::size_t>()->default_value(1),
                        "Number of io_context. If set 0 then automatically decided by hardware_concurrency()."
                )
                (
                        "threads_per_ioc",
                        boost::program_options::value<std::size_t>()->default_value(1),
                        "number of worker threads for each io_context."
                )
#if defined(MQTT_USE_LOG)
                (
                "verbose",
                boost::program_options::value<unsigned int>()->default_value(1),
                "set verbose level, possible values:\n 0 - Fatal\n 1 - Error\n 2 - Warning\n 3 - Info\n 4 - Debug\n 5 - Trace"
            )
#endif // defined(MQTT_USE_LOG)

#if defined(MQTT_USE_TLS)
                (
                "certificate",
                boost::program_options::value<std::string>(),
                "Certificate file for TLS connections"
            )
            (
                "private_key",
                boost::program_options::value<std::string>(),
                "Private key file for TLS connections"
            )
            (
                "verify_file",
                boost::program_options::value<std::string>(),
                "Root certificate used for verification of the client"
            )
            (
                "verify_field",
                boost::program_options::value<std::string>()->default_value("CN"),
                "Field to be used from certificate for authenticating clients"
            )
            (
                "certificate_reload_interval",
                boost::program_options::value<unsigned int>()->default_value(0),
                "Reload interval for the certificate and private key files (hours)\n 0 - Disabled"
            )
#endif
                (
                        "auth_file",
                        boost::program_options::value<std::string>(),
                        "Authentication file"
                )
                ;

        boost::program_options::options_description notls_desc("TCP Server options");
        notls_desc.add_options()
                ("tcp.port", boost::program_options::value<std::uint16_t>(), "default port (TCP)")
                ;
        desc.add(general_desc).add(notls_desc);

#if defined(MQTT_USE_WS)
        boost::program_options::options_description ws_desc("TCP websocket Server options");
        ws_desc.add_options()
            ("ws.port", boost::program_options::value<std::uint16_t>(), "default port (TCP)")
        ;

        desc.add(ws_desc);
#endif // defined(MQTT_USE_WS)

#if defined(MQTT_USE_TLS)
        boost::program_options::options_description tls_desc("TLS Server options");
        tls_desc.add_options()
            ("tls.port", boost::program_options::value<std::uint16_t>(), "default port (TLS)")
        ;

        desc.add(tls_desc);
#endif // defined(MQTT_USE_TLS)

#if defined(MQTT_USE_WS) && defined(MQTT_USE_TLS)
        boost::program_options::options_description tlsws_desc("TLS Websocket Server options");
        tlsws_desc.add_options()
            ("wss.port", boost::program_options::value<std::uint16_t>(), "default port (TLS)")
        ;
        desc.add(tlsws_desc);
#endif // defined(MQTT_USE_TLS) && defined(MQTT_USE_WS)

        boost::program_options::variables_map vm;
        boost::program_options::store(boost::program_options::parse_command_line(argc, argv, desc), vm);

        std::string config_file = vm["cfg"].as<std::string>();
        if (!config_file.empty()) {
            std::ifstream input(vm["cfg"].as<std::string>());
            if (input) {
                boost::program_options::store(boost::program_options::parse_config_file(input, desc), vm);
            }
            else {
                std::cerr
                        << "Configuration file '"
                        << config_file
                        << "' not found,  broker doesn't use configuration file."
                        << std::endl;
            }
        }

        boost::program_options::notify(vm);

        if (vm.count("help")) {
            std::cout << desc << std::endl;
            return 1;
        }

        std::cout << "Set options:" << std::endl;
        for (auto const& e : vm) {
            std::cout << boost::format("%-28s") % e.first.c_str() << " : ";
            if (auto p = boost::any_cast<std::string>(&e.second.value())) {
                std::cout << *p;
            }
            else if (auto p = boost::any_cast<std::size_t>(&e.second.value())) {
                std::cout << *p;
            }
            else if (auto p = boost::any_cast<std::uint32_t>(&e.second.value())) {
                std::cout << *p;
            }
            else if (auto p = boost::any_cast<std::uint16_t>(&e.second.value())) {
                std::cout << *p;
            }
            else if (auto p = boost::any_cast<unsigned int>(&e.second.value())) {
                std::cout << *p;
            }
            else if (auto p = boost::any_cast<bool>(&e.second.value())) {
                std::cout << std::boolalpha << *p;
            }
            std::cout << std::endl;
        }

#if defined(MQTT_USE_LOG)
        switch (vm["verbose"].as<unsigned int>()) {
        case 5:
            MQTT_NS::setup_log(MQTT_NS::severity_level::trace);
            break;
        case 4:
            MQTT_NS::setup_log(MQTT_NS::severity_level::debug);
            break;
        case 3:
            MQTT_NS::setup_log(MQTT_NS::severity_level::info);
            break;
        case 2:
            MQTT_NS::setup_log(MQTT_NS::severity_level::warning);
            break;
        default:
            MQTT_NS::setup_log(MQTT_NS::severity_level::error);
            break;
        case 0:
            MQTT_NS::setup_log(MQTT_NS::severity_level::fatal);
            break;
        }
#else
        MQTT_NS::setup_log();
#endif

        run_broker(vm);
    }
    catch(std::exception& e) {
        std::cerr << e.what() << std::endl;
    }
}*/

// header file //////////////////////////////////
#include <list>
#include <vector>
#include <boost/function.hpp>

class TimerManager;

class Timer
{
public:
    enum TimerType {ONCE, CIRCLE};

    explicit Timer(TimerManager& manager);
    ~Timer();

    template<typename Fun>
    void Start(Fun fun, unsigned interval, TimerType timeType = CIRCLE);
    void Stop();

private:
    void OnTimer(unsigned long long now);

private:
    friend class TimerManager;

    TimerManager& manager_;
    TimerType timerType_;
    boost::function<void(void)> timerFun_;
    unsigned interval_;
    unsigned long long expires_;

    int vecIndex_;
    std::list<Timer*>::iterator itr_;
};

class TimerManager
{
public:
    TimerManager();

    static unsigned long long GetCurrentMillisecs();
    void DetectTimers();

private:
    friend class Timer;
    void AddTimer(Timer* timer);
    void RemoveTimer(Timer* timer);

    int Cascade(int offset, int index);

private:
    typedef std::list<Timer*> TimeList;
    std::vector<TimeList> tvec_;
    unsigned long long checkTime_;
};

template<typename Fun>
inline void Timer::Start(Fun fun, unsigned interval, TimerType timeType)
{
    Stop();
    interval_ = interval;
    timerFun_ = fun;
    timerType_ = timeType;
    expires_ = interval_ + TimerManager::GetCurrentMillisecs();
    manager_.AddTimer(this);
}


// cpp file //////////////////////////////////////////////////

#ifdef _MSC_VER
# include <sys/timeb.h>
#else
# include <sys/time.h>
#endif

#define TVN_BITS 6
#define TVR_BITS 8
#define TVN_SIZE (1 << TVN_BITS)
#define TVR_SIZE (1 << TVR_BITS)
#define TVN_MASK (TVN_SIZE - 1)
#define TVR_MASK (TVR_SIZE - 1)
#define OFFSET(N) (TVR_SIZE + (N) *TVN_SIZE)
#define INDEX(V, N) ((V >> (TVR_BITS + (N) *TVN_BITS)) & TVN_MASK)

//////////////////////////////////////////////////////////////////////////
// Timer

Timer::Timer(TimerManager& manager)
        : manager_(manager)
        , vecIndex_(-1)
{
}

Timer::~Timer()
{
    Stop();
}

void Timer::Stop()
{
    if (vecIndex_ != -1)
    {
        manager_.RemoveTimer(this);
        vecIndex_ = -1;
    }
}

void Timer::OnTimer(unsigned long long now)
{
    if (timerType_ == Timer::CIRCLE)
    {
        expires_ = interval_ + now;
        manager_.AddTimer(this);
    }
    else
    {
        vecIndex_ = -1;
    }
    timerFun_();
}

//////////////////////////////////////////////////////////////////////////
// TimerManager

TimerManager::TimerManager()
{
    tvec_.resize(TVR_SIZE + 4 * TVN_SIZE);
    checkTime_ = GetCurrentMillisecs();
}

void TimerManager::AddTimer(Timer* timer)
{
    unsigned long long expires = timer->expires_;
    unsigned long long idx = expires - checkTime_;

    if (idx < TVR_SIZE)
    {
        timer->vecIndex_ = expires & TVR_MASK;
    }
    else if (idx < 1 << (TVR_BITS + TVN_BITS))
    {
        timer->vecIndex_ = OFFSET(0) + INDEX(expires, 0);
    }
    else if (idx < 1 << (TVR_BITS + 2 * TVN_BITS))
    {
        timer->vecIndex_ = OFFSET(1) + INDEX(expires, 1);
    }
    else if (idx < 1 << (TVR_BITS + 3 * TVN_BITS))
    {
        timer->vecIndex_ = OFFSET(2) + INDEX(expires, 2);
    }
    else if ((long long) idx < 0)
    {
        timer->vecIndex_ = checkTime_ & TVR_MASK;
    }
    else
    {
        if (idx > 0xffffffffUL)
        {
            idx = 0xffffffffUL;
            expires = idx + checkTime_;
        }
        timer->vecIndex_ = OFFSET(3) + INDEX(expires, 3);
    }

    TimeList& tlist = tvec_[timer->vecIndex_];
    tlist.push_back(timer);
    timer->itr_ = tlist.end();
    --timer->itr_;
}

void TimerManager::RemoveTimer(Timer* timer)
{
    TimeList& tlist = tvec_[timer->vecIndex_];
    tlist.erase(timer->itr_);
}

void TimerManager::DetectTimers()
{
    unsigned long long now = GetCurrentMillisecs();
    while (checkTime_ <= now)
    {
        int index = checkTime_ & TVR_MASK;
        if (!index &&
            !Cascade(OFFSET(0), INDEX(checkTime_, 0)) &&
            !Cascade(OFFSET(1), INDEX(checkTime_, 1)) &&
            !Cascade(OFFSET(2), INDEX(checkTime_, 2)))
        {
            Cascade(OFFSET(3), INDEX(checkTime_, 3));
        }
        ++checkTime_;

        TimeList& tlist = tvec_[index];
        TimeList temp;
        temp.splice(temp.end(), tlist);
        for (TimeList::iterator itr = temp.begin(); itr != temp.end(); ++itr)
        {
            (*itr)->OnTimer(now);
        }
    }
}

int TimerManager::Cascade(int offset, int index)
{
    TimeList& tlist = tvec_[offset + index];
    TimeList temp;
    temp.splice(temp.end(), tlist);

    for (TimeList::iterator itr = temp.begin(); itr != temp.end(); ++itr)
    {
        AddTimer(*itr);
    }

    return index;
}

unsigned long long TimerManager::GetCurrentMillisecs()
{
#ifdef _MSC_VER
    _timeb timebuffer;
    _ftime(&timebuffer);
    unsigned long long ret = timebuffer.time;
    ret = ret * 1000 + timebuffer.millitm;
    return ret;
#else
    timeval tv;
    ::gettimeofday(&tv, 0);
    unsigned long long ret = tv.tv_sec;
    return ret * 1000 + tv.tv_usec / 1000;
#endif
}